package com.qf.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 监听RabbitMQ的服务端
 *
 * @author lixu
 */
@Component //实例化
//@RabbitListener(queues = {"queue1","queue2"})
public class RabbitConfigListener {

    //分布式锁   setnx  :  setnx key value   判断key存在吗？ 如果存在就不设置了  如果不存在就设置
    @Autowired
    private RedisTemplate redisTemplate;

    //JUC 包下 java util c并发
    //本地锁
    //private final Lock lock = new ReentrantLock();

    /**
     * 接收避免重复消费的消息
     */
    @RabbitListener(queues = "QQ")
    public void receiverQQ(Object msg, Message message,Channel channel) throws Exception{
        //1：当前消费者  判断当前接收到的消息是否被消费了
        String msgId = (String) message.getMessageProperties().getHeaders()
                .get("spring_returned_message_correlation");

        Object value = redisTemplate.opsForValue().get(msgId);
        if ("0".equals(value)) {
            //3: 正在被消费中 id-0
            //锁：
            // 不见不散的锁   ： 必须获取锁  阻塞中此 就不走了
/*            lock.lock();// void    当前线程是一个状态 ：
            // 过期不候的锁  期： 就等1S
            boolean isLock = lock.tryLock(1, TimeUnit.SECONDS);
            // isLock:false 表示在1s之内没有获取到锁
            // isLock:true 表示在1s之内获取到锁
            if(isLock){
                //开始干活
            }else{
                //不能干活
            }*/
            //setIfAbsent : setnx
            Boolean isLock = redisTemplate.opsForValue()
                    .setIfAbsent("lock:" + msgId, "1", 2, TimeUnit.SECONDS);
            if(isLock){
                try {
                    //获取到了锁
                    System.out.println("接收到的消息:" + msg);
                    //..100行代码
                    //执行完成  改状态为1
                    redisTemplate.opsForValue().set(msgId,"1");
                    //正常应答
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                } finally {
                    //解锁
                    redisTemplate.delete("lock:" + msgId);
                }
            }else{
                //未获取到锁
                //我不干活 别人正在干活 、我也不能保存别人一定干成功  未应答
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
            }
            //  3.1: 尝试获取锁  锁获取到了  另一个消费者解锁了、并执行失败了
            //  3.2: 尝试获取锁  锁被占用   另一个消费者正在使用锁 、正在执行中   我也不干活我也不应答 basicNack
        } else {
            //2: 被消费了  id-1 直接手动应答即可
            System.out.println("消息已经消费完成");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    }

    //队列名称
    public static final String BOOT_TOPIC_QUEUE1 = "boot-queue1"; //常量
    public static final String BOOT_TOPIC_QUEUE2 = "boot-queue2"; //常量

    /**
     * 接收监听到的消息Q1   信道
     */
    @RabbitListener(queues = {"Q1"})
    public void receiverMessageQ1(String messageBody, Message message) {
        System.out.println("Q1消息体:" + messageBody);
    }

    /**
     * 接收监听到的消息Q2  信道
     */
    @RabbitListener(queues = {"Q2"})
    public void receiverMessageQ2(String messageBody, Message message) {
        System.out.println("Q2消息体:" + messageBody);
    }

    /**
     * 接收监听到的消息  信道
     */
    @RabbitListener(queues = {"queue1"})
    @RabbitHandler
    public void receiverMessage(String messageBody, Message message) {
        System.out.println("直接接收消息体:" + messageBody);
        System.out.println("接收消息对象：再获取消息体:" + new String(message.getBody()));
    }

    /**
     * 接收监听到的消息
     */
    @RabbitListener(queues = BOOT_TOPIC_QUEUE1)
    public void receiverBootTopicQueue1(String msg) {
        System.out.println(BOOT_TOPIC_QUEUE1 + "===" + msg);

    }

    /**
     * 接收监听到的消息
     */
/*    @RabbitListener(queues = BOOT_TOPIC_QUEUE2)
    public void receiverBootTopicQueue2(String msg){
        System.out.println(BOOT_TOPIC_QUEUE2 + "===" + msg);

    }*/

    /**
     * 接收监听到的消息
     * 应答： 手动应答
     * <p>
     * <p>
     * Message message:消息对象
     * 消息的属性 == 保存消息的相关属性  包含唯一标识
     * 消息体  == 保存消息的内容
     * Channel channel：信道
     */
    @RabbitListener(queues = BOOT_TOPIC_QUEUE2)
    public void receiverBootTopicQueueManual(String msg, Message message, Channel channel) {
        try {
            System.out.println(BOOT_TOPIC_QUEUE2 + "===" + msg);
            // 50行代码
            int j = 1 / 0;
            //手动应答
            //参数1: 应答的消费的唯一主键 （唯一标识）   --》5个消息
            //参数2： multiple 是否批量应答   false:不批量   true:批量应答      6,4,9,1,2
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {

            //判断当前消息是否投递过的
            if (message.getMessageProperties().isRedelivered()) {
                //是投递过的  第二次
                //转人工
                //将排队中的当前消息 移除   防止后续消费不能处理 造成消息的积压
                //System.out.println("不能接收成功的消息 记录下来");
                System.out.println("是投递过的  第二次 拒绝了");
                //参数1: 应答的消费的唯一主键 （唯一标识）   --》5个消息
                //参数2：重新放回队列    false: Unacked --> 移除了   true:Unacked --> 移动到 ready 中
                try {
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            } else {
                //没有投递过  第一次
                //手动不应答 表示没有处理好 再给一次机会
                //参数1: 应答的消费的唯一主键 （唯一标识）   --》5个消息
                //参数2： multiple 是否批量应答   false:不批量   true:批量应答      6,4,9,1,2
                //参数3：重新放回队列  Unacked --> 移动到 ready 中
                try {
                    System.out.println("没有投递过  第一次 给了一个机会");
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }

    }

}
